package com.cmsz.shuang.booter.rpc;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.atomic.AtomicInteger;
/**
 * RpcServer 类，采用ServerSocket启动服务端口， 客户端连接时，启动新的ProcessThread处理服务请求。
 * 			 通过 proxy类代理服务端的实现类。
 * @author shuangzh
 *
 */
public class RpcServer {
	private int port;	// 服务端口
	private Object actor;	// 被代理的类，此类实现了interfaceName指定的接口
	private String interfaceName;	// 接口类名
	private Proxy proxy;			// 代理对象
	private ServerSocket serverSocket = null;	// ServerSocket类，启动服务监听端口

	private volatile boolean isRunning;  // 运行状态控制
	private AtomicInteger threadCount = new AtomicInteger();

	private Thread mainThread = null;

	public RpcServer(String interfaceName, Object actor, int port) {
		this.interfaceName = interfaceName;
		this.actor = actor;
		this.port = port;
	}

	synchronized public void start() throws IOException, ClassNotFoundException {
		if (isRunning) {
			System.out.println("the rpc server is running");
			return;
		}
		// 设置代理对象
		proxy = new Proxy();
		proxy.setActor(actor);
		proxy.setInterfaceName(interfaceName);
		proxy.setClazz(Class.forName(interfaceName));
		
		// 启动 serversocket端口
		serverSocket = new ServerSocket(port);

		isRunning = true;
		mainThread = new Thread() {	// 接收 socket accept 的循环 线程类
			@Override
			public void run() {
				while (isRunning) {
					try {

						Socket s = serverSocket.accept();
						System.out.println("accept socket");
						ProcessThread pt = new ProcessThread(s);
						pt.setDaemon(true);
						pt.start();
					} catch (IOException e) {
						//e.printStackTrace();
					}
				}
			}
		};
		mainThread.setDaemon(false);
		mainThread.start();
	}

	synchronized public void stop() throws IOException {
		this.isRunning = false;
		if (serverSocket != null && (!serverSocket.isClosed())) {
			serverSocket.close();
		}
	}

	class ProcessThread extends Thread {
		Socket socket = null;
		Linker linker = null;

		ProcessThread(Socket s) throws IOException {
			this.socket = s;
			linker = new SocketLinker(s);
			super.setName("ProcessThread:" + threadCount.getAndIncrement());
		}

		@Override
		public void run() {
			System.out.println(Thread.currentThread().getName()
					+ " is running...");
			while (isRunning) {
				try {
					RpcRequest req = (RpcRequest) linker.receiveObject();
					System.out.println(Thread.currentThread().getName() + " handle " + req.getId() + " request");
					if (req.getArgs() == null) {
						req.setArgs(new Object[] {});
					}
					Object result = proxy.call(req.getMethodName(),
							req.getParameterTypes(), req.getArgs());

					RpcResponse rsp = new RpcResponse();
					rsp.setId(req.getId());
					rsp.setResult(result);
					linker.sendObject(rsp);

				} catch (Exception e) {
					// e.printStackTrace();
					break;
				}
			}
			System.out.println(Thread.currentThread().getName()
					+ " is stopping...");
			threadCount.decrementAndGet();
			try {
				if (socket.isConnected()) {
					socket.close();
				}
			} catch (IOException e) {
				e.printStackTrace();
			}
		}
	}

}